/*
Copyright (C) 2010- Peter Bui and Aaron Dingler.
This software is distributed under the GNU General Public License.
See the file COPYING for details.
*/

/* TODO: Synchronize timeouts between all programs & check time-to-live of messages*/
#include "smq_common.h"
#include "smq_catalog.h"
#include "smq_client.h"

#include <debug.h>
#include <link.h>

#include <string.h>
#include <stdlib.h>
#include <sys/stat.h>

static int SMQ_Client_Stalled_Timeout = SMQ_CLIENT_DEFAULT_STALLED_TIMEOUT;
static int SMQ_Client_Idle_Timeout    = SMQ_CLIENT_DEFAULT_IDLE_TIMEOUT;

struct link*
smq_client_connect(char *name, int check_name, int mode, int timeout)
{
    struct link *l = NULL;
    struct list *hl = NULL;
    struct smq_catalog_host *h = NULL;

    hl = smq_catalog_get_hosts(name, check_name);

    list_first_item(hl);
    if (mode == SMQ_CLIENT_CONNECT_RANDOM) {
	int i;
	int x = rand() % list_size(hl);

	list_first_item(hl);
	for (i = 0; i < x; i++)
	    h = list_next_item(hl);
    }

    while ((h = list_next_item(hl))) {
	debug(D_DEBUG, "connecting to %s:%d", h->address, h->port);
	l = link_connect(h->address, h->port, time(0) + timeout);
	if (l) break;
    }

    list_free(hl);
    list_delete(hl);
    return l;
}

int
smq_client_create(struct link *l, char *mq, INT64_T timeout)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "create \"%s\"\n", mq);
    debug(D_DEBUG, "create: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);
    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + timeout);

    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_put_from_buffer(struct link *l, char *target, INT64_T timeout, char *header_buffer, int hlen, char *body_buffer, int blen)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "put \"%s\" %lu %lu\n", target, (unsigned long)hlen, (unsigned long)blen);
    debug(D_DEBUG, "put_from_buffer: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);
    link_write(l, header_buffer, hlen, time(0) + SMQ_Client_Stalled_Timeout);
    link_write(l, body_buffer, blen, time(0) + SMQ_Client_Stalled_Timeout);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    printf("%s\n", rpc_buffer);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_put_from_file(struct link *l, char *target, INT64_T timeout, FILE* header_file, FILE* body_file)
{
    char rpc_buffer[SMQ_LINE_MAX];
    struct stat h_st;
    struct stat b_st;
    
    fstat(fileno(header_file), &h_st);
    fstat(fileno(body_file), &b_st);

    snprintf(rpc_buffer, SMQ_LINE_MAX, "put \"%s\" %ld %ld\n", target, h_st.st_size, b_st.st_size);
    debug(D_DEBUG, "put_from_file: %s", rpc_buffer);
    
    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);
    link_stream_from_file(l, header_file, h_st.st_size, time(0) + SMQ_Client_Stalled_Timeout);
    link_stream_from_file(l, body_file, b_st.st_size, time(0) + SMQ_Client_Stalled_Timeout);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    printf("%s\n", rpc_buffer);
    
    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_bind_from_buffer(struct link *l, char *target, INT64_T timeout, char *script_buffer, char* script_name, int slen)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "bind \"%s\" \"%s\" %lu\n", target, script_name, (unsigned long)slen);
    debug(D_DEBUG, "bind_from_buffer: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);
    link_write(l, script_buffer, strlen(script_buffer), time(0) + SMQ_Client_Stalled_Timeout);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_bind_from_file(struct link *l, char *target, INT64_T timeout, FILE* script_file, char *script_name)
{
    char rpc_buffer[SMQ_LINE_MAX];
    struct stat script_st;
    
    fstat(fileno(script_file), &script_st);

    snprintf(rpc_buffer, SMQ_LINE_MAX, "bind \"%s\" \"%s\" %ld\n", target, script_name, script_st.st_size);
    debug(D_DEBUG, "bind_from_file: %s", rpc_buffer);
    
    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);
    link_stream_from_file(l, script_file, script_st.st_size, time(0) + SMQ_Client_Stalled_Timeout);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_unbind(struct link *l, char *target, INT64_T timeout, char *script_name)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "unbind \"%s\" \"%s\"\n", target, script_name);
    debug(D_DEBUG, "unbind: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_bindings(struct link *l, char *target, INT64_T timeout)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "bindings \"%s\"\n", target);
    debug(D_DEBUG, "bindings: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);

    char rcv_buffer[SMQ_LINE_MAX];
    link_readline(l, rcv_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    char *p;
    char *tok = strdup(rcv_buffer);
    p = strtok(tok, " ");
    while(p != NULL)
    {
        printf("%s\n", p);
        p = strtok(NULL, " ");
    }

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_list(struct link *l, char *target, INT64_T timeout)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "list \"%s\"\n", target);
    debug(D_DEBUG, "list: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);

    char rcv_buffer[SMQ_LINE_MAX];
    link_readline(l, rcv_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    char *p;
    char *tok = strdup(rcv_buffer);
    p = strtok(tok, " ");
    while(p != NULL)
    {
        printf("%s\n", p);
        p = strtok(NULL, " ");
    }

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

int
smq_client_flush(struct link *l, char *target, INT64_T timeout)
{
    char rpc_buffer[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "flush \"%s\"\n", target);
    debug(D_DEBUG, "flush: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + timeout);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
}

//TODO: why doesn't this take timeout anymore?
int
smq_client_get(struct link* l, char *name, char *mid)
{
    char rpc_buffer[SMQ_LINE_MAX];
    int msglen, headerlen;
    char headername[SMQ_LINE_MAX];
    char msgname[SMQ_LINE_MAX];

    snprintf(rpc_buffer, SMQ_LINE_MAX, "get \"%s\" \"%s\"\n", name, mid);
    debug(D_DEBUG, "get: %s", rpc_buffer);

    link_write(l, rpc_buffer, strlen(rpc_buffer), time(0) + SMQ_Client_Stalled_Timeout);
    
    //get header length, name
    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    sscanf(rpc_buffer, "%d %s", &headerlen, headername);
    debug(D_DEBUG, "get header = %d %s", headerlen, headername);

    FILE *headerfile = fopen(headername, "w+");

    link_stream_to_file(l, headerfile, headerlen, time(0) + SMQ_Client_Stalled_Timeout);
    fclose(headerfile);
    
    //get message length, name
    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);

    sscanf(rpc_buffer, "%d %s", &msglen, msgname);
    debug(D_DEBUG, "get message = %d %s", msglen, msgname);

    FILE *msgfile = fopen(msgname, "w+");

    link_stream_to_file(l, msgfile, msglen, time(0) + SMQ_Client_Stalled_Timeout);
    fclose(msgfile);

    link_readline(l, rpc_buffer, SMQ_LINE_MAX, time(0) + SMQ_Client_Idle_Timeout);
    
    debug(D_DEBUG, "= %s", rpc_buffer);
    return strtoll(rpc_buffer, NULL, 10);
    
}

/* 
vim: sw=4 sts=4 ts=8 ft=cpp
*/
